/**
 * File: AbstractJobQueue.java
 * Created by: mhaimel
 * Created on: 31 Mar 2009
 * CVS:  $Id: AbstractJobQueue.java,v 1.5 2009/10/09 14:36:05 mhaimel Exp $
 */
package uk.ac.ebi.velvet.exec.impl;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import uk.ac.ebi.curtain.utils.concurrent.UtilsCollect;
import uk.ac.ebi.velvet.VelvetUncheckedException;

/**
 * @author mhaimel
 *
 */
public abstract class AbstractJobQueue <V,T extends Callable<V>,C> implements Callable<C>{
	private static final int _DEFAULT_SLEEP_TIME_MILLIS = (int) TimeUnit.SECONDS.toMillis(10);
	private final Log log = LogFactory.getLog(this.getClass());
	private final ConcurrentLinkedQueue<T> jobs = new ConcurrentLinkedQueue<T>();
	private final ConcurrentLinkedQueue<Future<V>> resList = new ConcurrentLinkedQueue<Future<V>>();
	private final ConcurrentMap<Future<V>, T> future2jobs = UtilsCollect.newConcurrentMap();
	private final AtomicBoolean shutDown = new AtomicBoolean(false);
	private final AtomicBoolean forceShutDown = new AtomicBoolean(false);
	private volatile Thread shutdownThread = null;
	private final ReentrantLock lock = new ReentrantLock();
	private final Condition submitCondition = lock.newCondition();
	
	public AbstractJobQueue() {
		// do nothing
	}
	
	public void submit(T job){
		lock.lock();
		try{
			this.jobs.add(job);
			submitCondition.signalAll();
		}finally{
			lock.unlock();
		}
	}
	
	private void registerShutdownhook() {
		Runtime.getRuntime().addShutdownHook(getShutdownThread());
	}
	
	private void removeShutdownhook() {
		Runtime.getRuntime().removeShutdownHook(getShutdownThread());
	}

	private Thread getShutdownThread() {
		if (null == shutdownThread){
			shutdownThread = new Thread() {
			    public void run() { 
			    	shutdown(true);
			    }
			};
		}
		return shutdownThread;
	}

	@Override
	public C call() throws Exception {
		registerShutdownhook();
		try{
			awaitStart();
			preCheck();
			while(keepProcessing()){
				while(hasMoreJobs()){
					processNewJobs(getNextJob());
				}
				if(!isForceShutDown()){
					checkResults();
				}
				sleep();
			}		
			if(isForceShutDown()){
				this.shutdown(true);
			}
			getLog().debug("Finished Processing job queue!");
			doShutdown();
			return getReturnCall();
		} finally{
			removeShutdownhook();
		}
	}

	private void awaitStart() {
		lock.lock();
		try{
			if(!_hasMoreJobs()){
				_awaitCondition(false);
			}
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}

	protected void preCheck() {
		// nothing to do
	}

	protected abstract C getReturnCall();

	private boolean hasMoreJobs() {
		
		return _hasMoreJobs();
	}

	private boolean _hasMoreJobs() {
		return !isForceShutDown() && !jobs.isEmpty();
	}

	private T getNextJob() {
		lock.lock();
		try{			
			return jobs.poll();	
		}finally{
			lock.unlock();
		}
	}

	protected abstract void doShutdown();

	protected void sleep() {
		lock.lock();
		try{
			if(!_isForceShutDown()){
				if(!_hasMoreJobs()){
					_awaitCondition(true);
				}
			}
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}

	private void _awaitCondition(boolean sleep) {
		try {
			if(sleep){
				submitCondition.await(sleepTime(),TimeUnit.MILLISECONDS);
			} else {
				submitCondition.await();
			}
		} catch (InterruptedException e) {
			getLog().debug("Thread interrupted!");
			Thread.currentThread().interrupt();
		}
	}

	protected int sleepTime() {
		return _DEFAULT_SLEEP_TIME_MILLIS;
	}

	protected void checkResults() {
		Future<V> nRes = null;
		// as long as there are finished results
		while((nRes = getNextResult())!= null){
			T job = future2jobs.remove(nRes);
			processResult(nRes,job);
		}
	}

	protected abstract void processResult(Future<V> res, T job);

	private Future<V> getNextResult() {
		Future<V> res = null;
		Future<V> firstRes = null;
		/* either one element is finished or the the whole queue got checked */
		while(!resList.isEmpty()){
			Future<V> nRes = resList.peek();
			if(null == firstRes){
				firstRes = nRes;
			} else {
				if(firstRes.equals(nRes)){
					// checked whole queue -> finish
					break;
				}
			}
			res = resList.poll();
			if(!res.equals(nRes)){
				throw new VelvetUncheckedException("Queue messed up peek and poll!");
			}
			if(nRes.isDone() || nRes.isCancelled()){
				break;
			} else {
				res = null;
			}
			_add2Future(nRes);
		}
		return res;
	}

	private void _add2Future(Future<V> res) {
		if(!resList.offer(res)){
			throw new VelvetUncheckedException("Failed to add element to Queue! " + res);
		}
	}

	private void processNewJobs(T job) {
		Future<V> res = submitJob(job);
		future2jobs.put(res, job);
		_add2Future(res);
	}

	protected abstract Future<V> submitJob(T job);

	protected boolean keepProcessing() {
		return !isForceShutDown() && (!isShutDown() || !resList.isEmpty() || hasMoreJobs());
	}

	private boolean isShutDown() {
		return shutDown.get();
	}
	
	public void shutdown(boolean force){
		boolean callForce = false;
		lock.lock();
		try{
			this.shutDown.set(true);
			if(force){
				if(!forceShutDown.getAndSet(true)){
					callForce = true;
				}
			}
			submitCondition.signalAll();
		}finally{
			lock.unlock();
		}
		if(callForce){
			forceShutdown();
		}
	}
	
	protected boolean isForceShutDown(){
		lock.lock();
		try{
			boolean isForce = _isForceShutDown();
			submitCondition.signalAll();
			return isForce;
		}finally{
			lock.unlock();
		}
	}

	private boolean _isForceShutDown() {
		return forceShutDown.get() || Thread.currentThread().isInterrupted();
	}

	protected abstract void forceShutdown();
	
	protected Log getLog() {
		return log;
	}
	
}
